package com.chencong.online.driver.dwd;

import com.chencong.online.bean.AppPromotionCountBean;
import com.chencong.online.bean.MarketingUserBehaviorBean;
import com.chencong.online.driver.ods.OdsMarketingUserBehaviorDriver;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.AppCountAggFunc;
import com.chencong.online.function.AppCountProcessWindowFunc;
import com.chencong.online.utils.TimestampTransformUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.Objects;

/**
 * @program: user-behavior-analysis-online
 * @ClassName DwdAppMarketingAnalysisDriver
 * @description:app市场分析
 * @author: chencong
 * @create: 2021-12-29 09:39
 **/
public class DwdAppMarketingAnalysisDriver {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
        env.setParallelism(1);

        //1、source
        DataStreamSource<MarketingUserBehaviorBean> inputDS = env.addSource(new OdsMarketingUserBehaviorDriver());
//        inputDS.print("初始化数据");

        //2、transform
        //过滤->提取时间戳作为事件时间
        SingleOutputStreamOperator<MarketingUserBehaviorBean> marketingUserBehaviorDS = inputDS
                .filter(data -> !Objects.equals(data.getBehavior(), "UNINSTALL"))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<MarketingUserBehaviorBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<MarketingUserBehaviorBean>() {
                            @Override
                            public long extractTimestamp(MarketingUserBehaviorBean element, long recordTimestamp) {
                                long milliTimestamp = TimestampTransformUtil.localDataTimeToMilliTimestamp(element.getLocalDateTime());
                                return milliTimestamp;
                            }
                        }));
//        marketingUserBehaviorDS.print("清洗后数据");

        //开窗求统计
        SingleOutputStreamOperator<AppPromotionCountBean> result = marketingUserBehaviorDS
                .keyBy("channel", "behavior")
                .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.seconds(10)))
                .aggregate(new AppCountAggFunc(), new AppCountProcessWindowFunc());

        result.print("结果为");
        env.execute("DwdAppMarketingAnalysisDriver");
    }
}
